package cn.doitedu.demo10;

import cn.doitedu.pojo.UserEvent;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.roaringbitmap.longlong.Roaring64Bitmap;
import redis.clients.jedis.Jedis;

import java.io.IOException;


/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2023/9/22
 * @Desc: 学大数据，上多易教育
 * 支持如下规则：
 * age>30的人，在2023-09-01到规则上线后，发生 C行为 3次 以上，当他做 V 事件则命中规则
 {
 "rule_id":"rule-3-001",
 "profile_condition":{
 "tag_name":"age",
 "tag_value":30
 },
 "realtime_profile":[
 {
 "start_time":"2023-09-01 08:00:00",
 "end_time": "2023-10-20 23:00:00",
 "event_id": "M",
 "min_count": 3,
 "rp_condition_id":1
 },
 {
 "start_time":"2023-09-11 08:00:00",
 "end_time": "",
 "event_id": "K",
 "min_count": 5,
 "rp_condition_id":2
 }
 ],
 "fire_event_id":"V"
 }
 **/
public class RuleModel_3_Calculator implements RuleCalculator {

    ValueState<Integer> cntState;
    JSONObject ruleParamObject;
    JSONObject message;
    Roaring64Bitmap preSelectUsersBitmap;
    Jedis jedis;
    String ruleId;
    Long webCalcEndTime;

    @Override
    public void init(RuntimeContext runtimeContext,
                     String ruleParamJson,
                     Roaring64Bitmap preSelectUsersBitmap,
                     Long webCalcEndTime) throws IOException {

        this.cntState = runtimeContext.getState(new ValueStateDescriptor<Integer>("cnt_state", Integer.class));
        this.ruleParamObject = JSON.parseObject(ruleParamJson);

        this.message = new JSONObject();
        this.ruleId = ruleParamObject.getString("rule_id");

        this.message.put("rule_id", ruleId);
        this.preSelectUsersBitmap = preSelectUsersBitmap;

        // 创建redis客户端连接
        this.jedis = new Jedis("doitedu", 6379);

        this.webCalcEndTime = webCalcEndTime;

    }

    @Override
    public void calc(UserEvent eventBean, Collector<String> out) throws IOException {

        // 判断收到的事件是否已经是过期的事件（不需要处理的历史事件）
        if (eventBean.getTimestamp() <= webCalcEndTime) return;

        // 先判断行为人是否属于本规则的目标人群
        if (!preSelectUsersBitmap.contains(eventBean.getUid())) return;

        /**
         * 动态画像
         */
        profileCalculate(eventBean);

        /**
         * 触发判断
         */
        fireJudge(eventBean, out);


    }

    /**
     * 触发后的条件判断
     *
     * @param eventBean
     * @param out
     * @throws IOException
     */
    private void fireJudge(UserEvent eventBean, Collector<String> out) throws IOException {
        String fireEventId = ruleParamObject.getString("fire_event_id");
        JSONObject realtimeProfile0 = ruleParamObject.getJSONArray("realtime_profile").getJSONObject(0);
        Integer minCount = realtimeProfile0.getInteger("min_count");


        // 判断动态画像条件是否满足
        if (eventBean.getEvent_id().equals(fireEventId)
                && cntState.value() != null
                && cntState.value() >= minCount
        ) {
            message.put("user_id", eventBean.getUid());
            message.put("hit_time", eventBean.getTimestamp());

            out.collect(message.toJSONString());

        }
    }

    /**
     * 动态画像统计
     *
     * @param eventBean
     * @throws IOException
     */
    private void profileCalculate(UserEvent eventBean) throws IOException {
        // 提取动态画像条件要统计的事件
        JSONArray realtimeProfile = ruleParamObject.getJSONArray("realtime_profile");

        // 取动态画像条件数组中的第0个条件
        JSONObject realtimeProfile_0 = realtimeProfile.getJSONObject(0);

        // 本条件的目标事件
        String targetEventId = realtimeProfile_0.getString("event_id");

        // 本条件的标识
        String rpCondition_id = realtimeProfile_0.getString("rp_condition_id");

        // 判断当前的事件，是否是动态画像的目标统计事件
        if (eventBean.getEvent_id().equals(targetEventId)) {

            // 先判断计数状态中是否有值，如果没有值，则去redis获取历史值
            Integer oldValue = cntState.value();

            if (oldValue == null) {
                // 查redis
                //  redis中的历史值数据存储结构：
                //      rule_id:条件id  ==> hash { user_id ==>  value }

                String redisValue = jedis.hget(ruleId + ":" + rpCondition_id, eventBean.getUid() + "");

                oldValue = redisValue == null ? 0 : Integer.parseInt(redisValue);

            }


            // 更新计数
            cntState.update(oldValue + 1);
        }
    }
}
